package com.feiwei

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.table.api.scala.{StreamTableEnvironment, _}
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Kafka, Schema}

object day4_TableTestFromKafka {



  def main(args: Array[String]): Unit = {
    val set = StreamExecutionEnvironment.getExecutionEnvironment

    val tableEnv = StreamTableEnvironment.create(set)

    //链接外部系统，创建表
    //2.1 读取文件


    tableEnv.connect(new Kafka().version("0.11")
                                 .topic("sensor")
      .property("zookeeper.connect", "localhost:2181")
      .property("bootstrap.servers", "localhost:9092")
    ).withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temperature", DataTypes.DOUBLE()))
      .createTemporaryTable("inputTable")


    val inputTable:Table=tableEnv.from("inputTable")

   // inputTable.toAppendStream[(String,Long, Double)].print()


    val t1: Table = inputTable
      .select('id, 'temperature)
      .filter('id === "sensor_1")


    var t2=tableEnv.sqlQuery(

      """
        |select id, temperature
        |from inputTable
        |where id = 'sensor_1'
        |
      """.stripMargin
    )
    t2.toAppendStream[(String, Double)].print()
    set.execute()
  }
}
